package com.navinfo.platform.examples.carbondata

import org.apache.hadoop.conf.Configuration

import org.apache.carbondata.hadoop.api.CarbonTableInputFormat
import org.apache.carbondata.hadoop.CarbonProjection

// scalastyle:off println
object HadoopFileExample {

  def main(args: Array[String]): Unit = {
    val spark = ExampleUtils.createCarbonSession("HadoopFileExample")
    ExampleUtils.writeSampleCarbonFile(spark, "carbon1")

    // read two columns
    val projection = new CarbonProjection
    projection.addColumn("c1")  // column c1
    projection.addColumn("c3")  // column c3
    val conf = new Configuration()
    CarbonTableInputFormat.setColumnProjection(conf, projection)
    CarbonTableInputFormat.setDatabaseName(conf, "default")
    CarbonTableInputFormat.setTableName(conf, "carbon1")

    val sc = spark.sparkContext
    val input = sc.newAPIHadoopFile(s"${ExampleUtils.storeLocation}/default/carbon1",
      classOf[CarbonTableInputFormat[Array[Object]]],
      classOf[Void],
      classOf[Array[Object]],
      conf)
    val result = input.map(x => x._2.toList).collect
    result.foreach(x => println(x.mkString(", ")))

    // delete carbondata file
    ExampleUtils.cleanSampleCarbonFile(spark, "carbon1")
  }
}
// scalastyle:on println
